package com.quarkus.aggregator.stream;

import com.quarkus.aggregator.model.Aggregation;
import com.quarkus.aggregator.model.TemperatureMeasurement;
import com.quarkus.aggregator.model.WeatherStation;
import io.quarkus.kafka.client.serialization.ObjectMapperSerde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.Stores;

import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.inject.Produces;
import java.time.Instant;

/**
 *  流处理 pipeline
 */
@ApplicationScoped
public class TopologyProducer {

    public static final String WEATHER_STATIONS_STORE = "weather-stations-store";

    private static final String WEATHER_STATIONS_TOPIC = "weather-stations";
    private static final String TEMPERATURE_VALUES_TOPIC = "temperature-values";
    private static final String TEMPERATURES_AGGREGATED_TOPIC = "temperatures-aggregated";

    @Produces
    public Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();
        // 序列化工具
        ObjectMapperSerde<WeatherStation> weatherStationSerde = new ObjectMapperSerde<>(WeatherStation.class);
        ObjectMapperSerde<Aggregation> aggregationSerde = new ObjectMapperSerde<>(Aggregation.class);
        // kafka 中状态存储的类
        KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(WEATHER_STATIONS_STORE);
        // 将WeatherStation的数据读入 GlobalKTable
        GlobalKTable<Integer, WeatherStation> stations = builder.globalTable( 
                                                                WEATHER_STATIONS_TOPIC,
                                                                Consumed.with(Serdes.Integer(), weatherStationSerde));
        // 将temperature-values主题下的每条数据作为一个stream进行处理
        builder.stream(
                        TEMPERATURE_VALUES_TOPIC,
                        Consumed.with(Serdes.Integer(), Serdes.String())
                )
                .join(                // 将气象站数据和温度数据结合
                        stations, // 气象站表
                        (stationId, timestampAndValue) -> stationId, // stationId 作为链接key
                        (timestampAndValue, station) -> {   // 具体的关联逻辑   timestampAndValue：温度值   station： 气象站信息
                            String[] parts = timestampAndValue.split(";");
                            return new TemperatureMeasurement(station.id, station.name,
                                    Instant.parse(parts[0]), Double.valueOf(parts[1]));
                        }
                )
                // 分组
                .groupByKey()
                // 聚合
                .aggregate(                                                   
                        Aggregation::new,  // 初始化聚合结果
                        // 聚合逻辑
                        (stationId, value, aggregation) -> aggregation.updateFrom(value),
                        Materialized.<Integer, Aggregation> as(storeSupplier)
                            .withKeySerde(Serdes.Integer())
                            .withValueSerde(aggregationSerde)
                )
                // 转成stream
                .toStream()
                // 发送到topic
                .to(                                                          
                        TEMPERATURES_AGGREGATED_TOPIC,
                        Produced.with(Serdes.Integer(), aggregationSerde)
                );

        return builder.build();
    }
}